/*
 * ads_mgr.cpp
 *
 *  Created on: Dec 21, 2012
 *      Author: wangmu
 */
#include "ads_test_mgr.h"
#include <sstream>
#include <stdlib.h>
#include <syslog.h>
#include <omp.h>
#include <time.h>

#include "csapp.h"
#include "../../../src/XMLInterface/xml_utils.h"
#define THREAD_NUMBER 5

#define XML_END_TAG 	"</Eadp>"
#define XML_END_TAG_LEN	7
#define XML_FILE_RECV_PATH 	"./testout/"
extern int make_socket_non_blocking (int sfd);

/**
 * used for initial the assist data
 * Every CAdsTestUser should copy this to it's own data space for thread safe
 * don't change the default value
 */
service_item_enable default_gps_service_list[MAX_GPS_SERVICE_NUMBER] = {
		{ADS_SERVICE_ALMANAC,false},
		{ADS_SERVICE_UTCModel,false},
		{ADS_SERVICE_IonosphericModel,false},
		{ADS_SERVICE_NavigationModel,false},
		{ADS_SERVICE_DGPSCorrections,false},
		{ADS_SERVICE_ReferenceTime,false},
		{ADS_SERVICE_RealTimeIntegrity,false}
};

unsigned int GetTickCount()
{
	struct timeval tv;
	if(gettimeofday(&tv,NULL) != 0)
		return 0;
	return (tv.tv_sec*1000) + (tv.tv_usec/1000);
}

/**
 * data_type_convert
 * data type convert to ads service type
 * (string to int)
 */
ads_service_type data_type_convert_to_int(const string &type_str)
{
	if(type_str == "Period")
		return ADS_SERVICE_PERIOD;
	if(type_str == "Once")
		return ADS_SERVICE_ONCE;
	if(type_str == "Onchange")
		return ADS_SERVICE_ONCHANGED;
	if(type_str == "Stop")
		return ADS_SERVICE_STOP;

	return ADS_SERVICE_UNSUPPORT;
}
/**
 *
 */
void data_type_convert_to_str(ads_service_type s_type, string& out)
{
	if(s_type == ADS_SERVICE_PERIOD)
		out = "Period";
	else if(s_type == ADS_SERVICE_ONCE)
		out = "Once";
	else if(s_type == ADS_SERVICE_ONCHANGED)
		out = "Onchange";
	else if(s_type == ADS_SERVICE_STOP)
		out = "Stop";
	else
		out = "Unsupport";
}
/**
 * service list used for send assist data with enable flag
 */

/**
 *
 *CAdsTestUser::CAdsTestUser
 */
CAdsTestUser::CAdsTestUser(const IUserObserver& ob)
:m_fd(-1),m_state(ADS_USER_INIT),m_seq_number(1),m_req_number(0),m_file_seq_number(0),m_observer(ob)
{
	Sem_init(&m_act_lck, 0, 1);      /* Binary semaphore for locking, every user will have one lock */
	memcpy(m_gps_srv_en_lst,default_gps_service_list,sizeof(default_gps_service_list));
}

/**
 * CAdsTestUser::~CAdsTestUser()
 */
CAdsTestUser::~CAdsTestUser()
{
	clear_state();
}
/*
 * send_xml to socket
 *
 */
int CAdsTestUser::send_xml_data(const string& xml_data)
{
	int ret = 0;
	if(m_fd > 0)
	{
		ret = Rio_writen(m_fd,(void*) xml_data.c_str(),xml_data.size());         //line:netp:servestatic:write
		return ret;
	}else return -1;
}
/**
 * send_xml_file
 * return -2: file not exist
 * return -1: socket error
 * return > 0: success
 */
int CAdsTestUser::send_xml_file(const string& xml_file)
{
	struct stat sbuf;
	char *srcp;
	if (stat(xml_file.c_str(), &sbuf) < 0) {                     //line:netp:doit:beginnotfound
		printf("Forbidden Tiny couldn't read the file");
		return 0;
	}                                                    //line:netp:doit:endnotfound

	if (!(S_ISREG(sbuf.st_mode)) || !(S_IRUSR & sbuf.st_mode)) { //line:netp:doit:readable
		printf("Forbidden Tiny couldn't read the file");
		return 0;
	}
	syslog(LOG_INFO,"%s: send file %s, file size %d on fd %d\n",__PRETTY_FUNCTION__,xml_file.c_str(),sbuf.st_size, m_fd);
	int srcfd;
	/* Send response body to client */
	srcfd = Open(xml_file.c_str(), O_RDONLY, 0);    //line:netp:servestatic:open
	srcp = (char*) Mmap(0, sbuf.st_size, PROT_READ, MAP_PRIVATE, srcfd, 0);//line:netp:servestatic:mmap
	Close(srcfd);                           //line:netp:servestatic:close
	int ret = Rio_writen(m_fd, srcp, sbuf.st_size);         //line:netp:servestatic:write
	Munmap(srcp, sbuf.st_size);                 //line:netp:servestatic:munmap
	m_sent_bytes += sbuf.st_size;
	return ret;

}

/**
 * CAdsTestUser::read_socket(string &xmlbuffer)
 * success: return > recv size
 * failed: -1 , if failed, close (fd);
 */
int CAdsTestUser::read_socket(string &xmlbuffer)
{
	ssize_t n = 0;
	ssize_t nread = 0;
	char buf[BUFSIZ];
	while ((nread = read(m_fd, buf, BUFSIZ-1)) > 0) {
		n += nread;
		buf[nread] = 0;
		xmlbuffer+=buf;
		memset(buf,0,sizeof(buf));
		//printf("total read count %d, this time cout %d\n",n,nread);
	}
	if (nread == -1 && errno != EAGAIN) {
		perror("read error");
		return -1;
	}else
		if(nread == 0)
		{
			perror("remote closed socket");
			return -1;
		}
	syslog(LOG_INFO,"%s xml recv size %d on fd %d\n",__PRETTY_FUNCTION__,xmlbuffer.size(), m_fd);
	return xmlbuffer.size();
}
/**
 * main do ads action
 * if some error happen(read/write socket, parser xml, check parameter)
 * 	  will close socket, clear state, save state to old_list for future use
 */
int CAdsTestUser::do_ads_action()
{
	get_lck();
	int ret = -1;
	syslog(LOG_INFO,"%s user [%d] at state %d\n",__PRETTY_FUNCTION__, m_fd, m_state);
	if(m_fd < 0 || m_state < ADS_USER_CONNECTED)
	{
		release_lck();
		return -1;
	}
	string xmlstr_recv; // will be used for parser and create res
	if(-1 == read_socket(xmlstr_recv))
	{
		clear_state();
		release_lck();
		return -1;
	}

	//take care the whole xml can't recv in one time
	m_recv += xmlstr_recv;
	while(1)
	{
		size_t pos = 0;
		if((pos = m_recv.find(XML_END_TAG)) == string::npos )
		{// not found the XML_END_TAG, continue
			release_lck();
			return 0;
		}else{
			size_t spos = pos+XML_END_TAG_LEN;
			xmlstr_recv = m_recv.substr(0,spos);//
			string leftstr = m_recv.substr(spos);
			if(spos == m_recv.size())
				m_recv.clear();
			else
				m_recv = leftstr;
			m_recv_bytes += xmlstr_recv.size();
			switch(m_state)
			{
			case ADS_USER_SENT_SUB://xml should be subscribe response
				save_xml_data_file(xmlstr_recv,"subscribe_resp");
				m_state = ADS_USER_RECV_SUB_RESP;
				break;
			case ADS_USER_SENT_DATA_REQ: // xml should be data_req resp
				save_xml_data_file(xmlstr_recv,"data_req_resp");
				m_state = ADS_USER_RECV_DATA_REQ_RESP;
				break;
			case ADS_USER_RECV_DATA_REQ_RESP:
			case ADS_USER_RECV_ASSIST_DATA:
				save_xml_data_file(xmlstr_recv,"assist_data");
				break;
			case ADS_USER_CONNECTED:			// fd connected
			case ADS_USER_RECV_SUB_RESP:
			case ADS_USER_SENT_UN_SUB:
			default:
				syslog(LOG_INFO,"%s recv xml received in wrong state\n",__PRETTY_FUNCTION__);
				save_xml_data_file(xmlstr_recv,"wrong_state");
				break;
			}

		}
	}
	/*
	 * check ret value
	 * if error do action
	 */
	if(ret == -1)
	{
		clear_state();
	}
	release_lck();
	return ret;
}

/**
 * save xml file to folder XML_FILE_RECV_PATH
 * format:XML_FILE_RECV_PATH/xml_file_on_fd_[]_file_seq_[].xml
 */
int CAdsTestUser::save_xml_data_file(const string &xml_data, const string &xml_file)
{
	string new_xml_filename = XML_FILE_RECV_PATH;
	new_xml_filename += xml_file;
	new_xml_filename += "_on_fd_";
	new_xml_filename += str_convert<string>(m_fd);
	new_xml_filename += "_file_seq_";
	new_xml_filename += str_convert<string>(m_file_seq_number++);
	new_xml_filename += ".xml";
	ofstream out;
	out.open(new_xml_filename.c_str());
	if (out.is_open())
	{
		out << xml_data;
		out.close();
	}
	return 0;
}
/*
 * clear_state
 */
void CAdsTestUser::clear_state()
{
	syslog(LOG_INFO,"%s %d\n",__PRETTY_FUNCTION__,m_fd);
	if(m_fd > 0)
		Close(m_fd);
	m_fd = -1;
	m_sent_bytes = 0;
	m_recv_bytes = 0;
	m_state =ADS_USER_INIT;
	m_seq_number = 0;
	m_req_number = 0;
	m_services_list.clear();
	m_ref_l.ref_altitude.clear();
	m_ref_l.ref_latitude.clear();
	m_ref_l.ref_longitude.clear();
	m_user_id.clear();
	m_user_pswd.clear();
	m_recv.clear();
	memcpy(m_gps_srv_en_lst,default_gps_service_list,sizeof(default_gps_service_list));
}

/**
 *
 */
CUserMgr::CUserMgr(string host,int port, int max_users):
	m_host(host), m_port(port), m_max_user(max_users),m_total_fd(0), m_epoll_fd(-1)
{
	Sem_init(&m_on_line_lck, 0, 1);      /* Binary semaphore for locking, every user will have one lock */
	m_epoll_fd = epoll_create1(0);
	if (m_epoll_fd == -1)
	{
		perror ("epoll_create1");
		abort ();
	}
	struct epoll_event ev;
	ev.data.fd = STDIN_FILENO;
	ev.events = EPOLLIN|EPOLLET;
	int s=epoll_ctl(m_epoll_fd,EPOLL_CTL_ADD,STDIN_FILENO,&ev);
	if (s == -1)
	{
		perror ("epoll_ctl STDIN_FILENO");
		abort ();
	}

	for( int i = 0; i < max_users; i++)
	{
		ads_ptr temp(new CAdsTestUser(*this));
		m_on_line_users.push_back(temp);
	}
	//printf("m_on_line_user %d\n",m_on_line_users.size());
}
/**
 * create new connection
 */
int CUserMgr::create_connection(int cnn_number)
{
	//create connection with the server
	//add to epoll if connect failed success
	//abort if connect failed
	syslog(LOG_INFO,"%s  %d\n", __PRETTY_FUNCTION__,cnn_number);
	omp_set_num_threads(THREAD_NUMBER);
#pragma omp parallel for
	for(int i = 0; i < cnn_number;i++)
	{
		//printf("i = %d, thread num = %d\n", i,omp_get_thread_num());
		//clientfd will be unique for the exe, so this will
		//be used for find empty slot in g_user_mgr
		int clientfd = Open_clientfd((char*)m_host.c_str(), m_port);
		int s;
		struct epoll_event ev;

		if(clientfd == -1)
		{
			perror ("Open_clientfd error, stop create connection\n");
			syslog(LOG_INFO,"%s Open_clientfd error, stop create connection\n",__PRETTY_FUNCTION__);
			//abort ();
		}else
		{
			s = make_socket_non_blocking (clientfd);
			if (s == -1)
			{
				perror ("make_socket_non_blocking");
				abort ();
			}
			// Add to epoll
			ev.data.fd = clientfd;
			ev.events = EPOLLIN|EPOLLET;
			s = epoll_ctl(m_epoll_fd,EPOLL_CTL_ADD,clientfd,&ev);
			if (s == -1)
			{
				perror ("epoll_ctl clientfd");
				abort ();
			}
			ads_ptr user;
			if((user = get_user(clientfd)) != NULL)
			{
				user->init_user();
				user->set_fd(clientfd);
				user->set_user_state(ADS_USER_CONNECTED);
			}else {
				perror ("beyond the max capacity,close the socket!\n");
				Close(clientfd);
			}
		}
	}

	return cnn_number;
}

ads_ptr CUserMgr::get_user(int fd){
	P(&m_on_line_lck);
	ads_ptr temp;
	if(fd >= m_on_line_users.size())
	{
		syslog(LOG_INFO,"%s new m_online_user on %d, m_on_line_user.size %d\n", __PRETTY_FUNCTION__,fd,m_on_line_users.size());

		int i = m_on_line_users.size();
		while(i <= fd)
		{
			ads_ptr temp(new CAdsTestUser(*this));
			m_on_line_users.push_back(temp);
			i++;
		}
	}
	V(&m_on_line_lck);
	return m_on_line_users[fd];
}

int CUserMgr::show_statistics(bool b_detail)
{
	vector_ads_type::iterator user_ptr;
	int online = 0;
	for(user_ptr = m_on_line_users.begin(); user_ptr!=m_on_line_users.end(); user_ptr++)
	{
		if((*user_ptr)->get_state() >= ADS_USER_CONNECTED)
		{
			online++;
			if(b_detail)
			{
				printf("fd %d at state %s, send bytes %d, recv bytes %d\n",
						(*user_ptr)->get_fd(), USER_STATE_STR((*user_ptr)->get_state()),
						(*user_ptr)->get_send_bytes(), (*user_ptr)->get_recv_bytes());
			}

		}
	}
	printf("online connection: %d\n\n", online);

}

int CUserMgr::send_xml_and_change_state(const string &filename, User_State new_state)
{
	int start = GetTickCount();
	omp_set_num_threads(THREAD_NUMBER);
#pragma omp parallel for
	for(size_t i = 0; i < m_on_line_users.size();i++)
	{
		//printf("fd = %d,i %d, thread num = %d\n", m_on_line_users[i]->get_fd(),i,omp_get_thread_num());

		if(m_on_line_users[i]->get_state() >= ADS_USER_CONNECTED)
		{
			m_on_line_users[i]->get_lck();
			int ret = m_on_line_users[i]->send_xml_file(filename);
			if( -1 == ret)
			{
				syslog(LOG_ERR,"%s error %d on socket %d, reset all data\n", __PRETTY_FUNCTION__,errno, m_on_line_users[i]->get_fd());
				m_on_line_users[i]->init_user();

			}else if ( -2 == ret)
			{
				printf("file not exist: %s\n", filename.c_str());
			}
			else
			{
				m_on_line_users[i]->set_user_state(new_state);
			}
			m_on_line_users[i]->release_lck();
		}
	}
	printf("%d time used\n",GetTickCount()-start);
	return 0;
}
